package tom.spring.cloud.stream.reactive.sender;

import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.reactive.FluxSender;
import org.springframework.cloud.stream.reactive.StreamEmitter;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import tom.spring.cloud.msg.MyInput;
import tom.spring.cloud.msg.MyOutput;

import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * reactive 生产者
 * @author ZHUFEIFEI
 */
@Slf4j
@EnableBinding({MyOutput.class, MyInput.class})
public class RxProducer {

    /**
     * 响应式发送StreamEmitter
     * 方式1
     * @return
     */
    @StreamEmitter
    @Output(MyOutput.OUTPUT)
    public Flux<Date> emit010() {
        return Flux.interval(Duration.ofSeconds(2)).map(i -> {Date date = new Date();
        log.info("emit010 : {}", date); return date;});
    }

    /**
     * 方式2
     * @param output
     */
    @StreamEmitter
    @Output(MyOutput.OUTPUT)
    public void emit011(FluxSender output) {
        output.send(Flux.interval(Duration.ofSeconds(2)).map(i -> {Date date = new Date();
            log.info("emit011 : {}", date); return date;}));
    }

    /**
     * 方式3
     * @param output
     */
    @StreamEmitter
    public void emit012(@Output(MyOutput.OUTPUT) FluxSender output) {
        output.send(Flux.interval(Duration.ofSeconds(2)).map(i -> {Date date = new Date();
            log.info("emit012 : {}", date); return date;}));
    }

    /**
     * 方式4
     * @return
     */
    @Bean
    @StreamEmitter
    @Output(MyOutput.OUTPUT)
    public Publisher<Message<Date>> emit013() {
        return IntegrationFlows.from(
                () -> {Date date = new Date();log.info("emit013 : {}", date); return MessageBuilder.withPayload(date).build();},
                e -> e.poller(p -> p.fixedRate(2, TimeUnit.SECONDS))
        ).toReactivePublisher();
    }

    /**
     * 监听反馈, INPUT对应的channel为output_1，消费者反馈信息到该通道
     * @param flux
     */
    @StreamListener
    public void feedback(@Input(MyInput.INPUT) Flux<String> flux) {
        flux.subscribe(m -> log.info("feedback : {}", m));
    }

}
